{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Transforming and joining raw data\n",
    "\n",
    "The \"raw\" data is divided among the following tables:\n",
    "\n",
    "- **Customer metadata**\n",
    "  - customerID\n",
    "  - gender\n",
    "  - date of birth (we'll derive age and senior citizen status from this)\n",
    "  - Partner\n",
    "  - Dependents\n",
    "  - (nominal) MonthlyCharges\n",
    "- **Billing events**\n",
    "  - customerID\n",
    "  - date (we'll derive tenure from the number/duration of billing events)\n",
    "  - kind (one of \"AccountCreation\", \"Charge\", or \"AccountTermination\")\n",
    "  - value (either a positive nonzero amount or 0.00; we'll derive TotalCharges from the sum of amounts and Churn from the existence of an AccountTermination event)\n",
    "- **Customer phone features**\n",
    "  - customerID\n",
    "  - feature (one of \"PhoneService\" or \"MultipleLines\")\n",
    "- **Customer internet features**\n",
    "  - customerID\n",
    "  - feature (one of \"InternetService\", \"OnlineSecurity\", \"OnlineBackup\", \"DeviceProtection\", \"TechSupport\", \"StreamingTV\", \"StreamingMovies\")\n",
    "  - value (one of \"Fiber\", \"DSL\", \"Yes\", \"No\")\n",
    "- **Customer account features**\n",
    "  - customerID\n",
    "  - feature (one of \"Contract\", \"PaperlessBilling\", \"PaymentMethod\")\n",
    "  - value (one of \"Month-to-month\", \"One year\", \"Two year\", \"No\", \"Yes\", \"Credit card (automatic)\", \"Mailed check\", \"Bank transfer (automatic)\", \"Electronic check\")\n",
    "\n",
    "We want to join these together to reconstitute a training data set with this schema:\n",
    "\n",
    "- customerID\n",
    "- gender\n",
    "- SeniorCitizen\n",
    "- Partner\n",
    "- Dependents\n",
    "- tenure\n",
    "- PhoneService\n",
    "- MultipleLines\n",
    "- InternetService\n",
    "- OnlineSecurity\n",
    "- OnlineBackup\n",
    "- DeviceProtection\n",
    "- TechSupport\n",
    "- StreamingTV\n",
    "- StreamingMovies\n",
    "- Contract\n",
    "- PaperlessBilling\n",
    "- PaymentMethod\n",
    "- MonthlyCharges\n",
    "- TotalCharges\n",
    "- Churn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "tags": [
     "parameters"
    ]
   },
   "outputs": [],
   "source": [
    "# notebook parameters\n",
    "\n",
    "import os\n",
    "\n",
    "spark_master = os.getenv(\"SPARK_MASTER_URL\", \"/spark://ip:port\")\n",
    "app_name = \"churn-etl\"\n",
    "input_files = dict(\n",
    "    billing=\"billing_events\", \n",
    "    account_features=\"customer_account_features\", \n",
    "    internet_features=\"customer_internet_features\", \n",
    "    meta=\"customer_meta\", \n",
    "    phone_features=\"customer_phone_features\"\n",
    ")\n",
    "output_file = \"churn-etl\"\n",
    "output_mode = \"overwrite\"\n",
    "output_kind = \"parquet\"\n",
    "input_kind = \"parquet\"\n",
    "driver_memory = '8g'\n",
    "executor_memory = '8g'\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - hive</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://10.19.183.210:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.2.0</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>spark://yuanli-System-Product-Name:7077</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>PySparkShell</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7efe7c3e15b0>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import pyspark\n",
    "\n",
    "session = pyspark.sql.SparkSession.builder \\\n",
    "    .master(spark_master) \\\n",
    "    .appName(app_name) \\\n",
    "    .config(\"spark.eventLog.enabled\", True) \\\n",
    "    .config(\"spark.eventLog.dir\", \".\") \\\n",
    "    .config(\"spark.driver.memory\", driver_memory) \\\n",
    "    .config(\"spark.executor.memory\", executor_memory) \\\n",
    "    .getOrCreate()\n",
    "session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "import churn.etl\n",
    "\n",
    "churn.etl.register_options(\n",
    "    spark_master = spark_master,\n",
    "    app_name = app_name,\n",
    "    input_files = input_files,\n",
    "    output_mode = output_mode,\n",
    "    output_kind = output_kind,\n",
    "    input_kind = input_kind,\n",
    "    driver_memory = driver_memory,\n",
    "    executor_memory = executor_memory\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Reconstructing billing events and charges"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- customerID: string (nullable = true)\n",
      " |-- kind: string (nullable = true)\n",
      " |-- value: decimal(8,2) (nullable = true)\n",
      " |-- date: date (nullable = true)\n",
      " |-- month: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from churn.etl import read_df\n",
    "billing_events = read_df(session, input_files[\"billing\"])\n",
    "billing_events.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "from churn.etl import join_billing_data\n",
    "customer_billing = join_billing_data(billing_events)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[customerID: string, Churn: boolean, tenure: bigint, TotalCharges: decimal(18,2)]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "customer_billing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "When we aggregated billing data, we also captured a unique list of customers in a temporary view.  For convenience, we can access it as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "from churn.etl import customers as get_customers\n",
    "customers = get_customers()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Reconstructing phone features\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- customerID: string (nullable = true)\n",
      " |-- feature: string (nullable = true)\n",
      " |-- value: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "phone_features = read_df(session, input_files[\"phone_features\"])\n",
    "phone_features.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "from churn.etl import join_phone_features\n",
    "customer_phone_features = join_phone_features(phone_features)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Reconstructing internet features\n",
    "\n",
    "Whereas phone features only include whether or not there are multiple lines, there are several internet-specific features in accounts:\n",
    "\n",
    "- `InternetService` (one of `Fiber optic` or `DSL` in the \"raw\" data; its absence translates to `No` in the processed data)\n",
    "- `OnlineSecurity` (`Yes` in the \"raw\" data if present; one of `No`, `Yes`, or `No internet service` in the processed data)\n",
    "- `OnlineBackup` (`Yes` in the \"raw\" data if present; one of `No`, `Yes`, or `No internet service` in the processed data)\n",
    "- `DeviceProtection` (`Yes` in the \"raw\" data if present; one of `No`, `Yes`, or `No internet service` in the processed data)\n",
    "- `TechSupport` (`Yes` in the \"raw\" data if present; one of `No`, `Yes`, or `No internet service` in the processed data)\n",
    "- `StreamingTV` (`Yes` in the \"raw\" data if present; one of `No`, `Yes`, or `No internet service` in the processed data)\n",
    "- `StreamingMovies` (`Yes` in the \"raw\" data if present; one of `No`, `Yes`, or `No internet service` in the processed data)\n",
    "\n",
    "This will lead to some slightly more interesting joins!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- customerID: string (nullable = true)\n",
      " |-- feature: string (nullable = true)\n",
      " |-- value: string (nullable = true)\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2022-04-05 09:59:39,224 WARN rapids.GpuOverrides: \n",
      "!Exec <CollectLimitExec> cannot run on GPU because the Exec CollectLimitExec has been disabled, and is disabled by default because Collect Limit replacement can be slower on the GPU, if huge number of rows in a batch it could help by limiting the number of rows transferred from GPU to CPU. Set spark.rapids.sql.exec.CollectLimitExec to true if you wish to enable it\n",
      "  @Partitioning <SinglePartition$> could run on GPU\n",
      "\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+---------------+-----+\n",
      "|          customerID|        feature|value|\n",
      "+--------------------+---------------+-----+\n",
      "|7590-VHVEG-Mg8VG5...|InternetService|  DSL|\n",
      "|7590-VHVEG-5xLi5Z...|InternetService|  DSL|\n",
      "|7590-VHVEG-ZePlJi...|InternetService|  DSL|\n",
      "|7590-VHVEG-x9IoNd...|InternetService|  DSL|\n",
      "|7590-VHVEG-Z9yCIk...|InternetService|  DSL|\n",
      "|7590-VHVEG-K8kBya...|InternetService|  DSL|\n",
      "|7590-VHVEG-4ZjnIU...|InternetService|  DSL|\n",
      "|7590-VHVEG-0stTDJ...|InternetService|  DSL|\n",
      "|7590-VHVEG-lqhKlh...|InternetService|  DSL|\n",
      "|7590-VHVEG-4Y_zUA...|InternetService|  DSL|\n",
      "|7590-VHVEG-34V86Q...|InternetService|  DSL|\n",
      "|7590-VHVEG-GCNzU2...|InternetService|  DSL|\n",
      "|7590-VHVEG-i0AFUE...|InternetService|  DSL|\n",
      "|7590-VHVEG-F1ALBc...|InternetService|  DSL|\n",
      "|7590-VHVEG-aEfHl7...|InternetService|  DSL|\n",
      "|7590-VHVEG-eiqTDe...|InternetService|  DSL|\n",
      "|7590-VHVEG-3K15yQ...|InternetService|  DSL|\n",
      "|7590-VHVEG-iMYyeZ...|InternetService|  DSL|\n",
      "|7590-VHVEG-rReekB...|InternetService|  DSL|\n",
      "|7590-VHVEG-2l92Zs...|InternetService|  DSL|\n",
      "+--------------------+---------------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "internet_features = read_df(session, input_files[\"internet_features\"])\n",
    "internet_features.printSchema()\n",
    "internet_features.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "from churn.etl import join_internet_features\n",
    "customer_internet_features = join_internet_features(internet_features)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Reconstructing account features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- customerID: string (nullable = true)\n",
      " |-- feature: string (nullable = true)\n",
      " |-- value: string (nullable = true)\n",
      "\n",
      "+--------------------+-------------+----------------+\n",
      "|          customerID|      feature|           value|\n",
      "+--------------------+-------------+----------------+\n",
      "|7590-VHVEG-Mg8VG5...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-5xLi5Z...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-ZePlJi...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-x9IoNd...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-Z9yCIk...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-K8kBya...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-4ZjnIU...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-0stTDJ...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-lqhKlh...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-4Y_zUA...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-34V86Q...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-GCNzU2...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-i0AFUE...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-F1ALBc...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-aEfHl7...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-3K15yQ...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-eiqTDe...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-iMYyeZ...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-rReekB...|PaymentMethod|Electronic check|\n",
      "|7590-VHVEG-2l92Zs...|PaymentMethod|Electronic check|\n",
      "+--------------------+-------------+----------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2022-04-05 09:59:42,068 WARN rapids.GpuOverrides: \n",
      "!Exec <CollectLimitExec> cannot run on GPU because the Exec CollectLimitExec has been disabled, and is disabled by default because Collect Limit replacement can be slower on the GPU, if huge number of rows in a batch it could help by limiting the number of rows transferred from GPU to CPU. Set spark.rapids.sql.exec.CollectLimitExec to true if you wish to enable it\n",
      "  @Partitioning <SinglePartition$> could run on GPU\n",
      "\n"
     ]
    }
   ],
   "source": [
    "account_features = read_df(session, input_files[\"account_features\"])\n",
    "account_features.printSchema()\n",
    "account_features.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "from churn.etl import join_account_features\n",
    "customer_account_features = join_account_features(account_features)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Account metadata"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- customerID: string (nullable = true)\n",
      " |-- dateOfBirth: date (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- SeniorCitizen: string (nullable = true)\n",
      " |-- Partner: string (nullable = true)\n",
      " |-- Dependents: string (nullable = true)\n",
      " |-- MonthlyCharges: decimal(8,2) (nullable = true)\n",
      " |-- now: timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "account_meta = read_df(session, input_files[\"meta\"])\n",
    "account_meta.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "from churn.etl import process_account_meta\n",
    "customer_account_meta = process_account_meta(account_meta)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Putting it all together"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "from churn.etl import chained_join\n",
    "from churn.etl import forcefloat\n",
    "\n",
    "wide_data = chained_join(\n",
    "    \"customerID\",\n",
    "    customers,\n",
    "    [\n",
    "        customer_billing,\n",
    "        customer_phone_features,\n",
    "        customer_internet_features,\n",
    "        customer_account_features,\n",
    "        customer_account_meta\n",
    "    ]\n",
    ").select(\n",
    "    \"customerID\", \n",
    "    \"gender\", \n",
    "    \"SeniorCitizen\", \n",
    "    \"Partner\", \n",
    "    \"Dependents\", \n",
    "    \"tenure\", \n",
    "    \"PhoneService\", \n",
    "    \"MultipleLines\", \n",
    "    \"InternetService\", \n",
    "    \"OnlineSecurity\", \n",
    "    \"OnlineBackup\", \n",
    "    \"DeviceProtection\", \n",
    "    \"TechSupport\", \n",
    "    \"StreamingTV\", \n",
    "    \"StreamingMovies\", \n",
    "    \"Contract\", \n",
    "    \"PaperlessBilling\", \n",
    "    \"PaymentMethod\", \n",
    "    forcefloat(\"MonthlyCharges\"),\n",
    "    forcefloat(\"TotalCharges\"), \n",
    "    \"Churn\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "== Physical Plan ==\n",
      "AdaptiveSparkPlan isFinalPlan=false\n",
      "+- Project [customerID#0, gender#265, SeniorCitizen#279, Partner#267, Dependents#268, tenure#61L, PhoneService#97, MultipleLines#101, InternetService#199, OnlineSecurity#200, OnlineBackup#201, DeviceProtection#202, TechSupport#203, StreamingTV#204, StreamingMovies#205, Contract#233, PaperlessBilling#258, PaymentMethod#239, cast(MonthlyCharges#269 as float) AS MonthlyCharges#366, cast(TotalCharges#62 as float) AS TotalCharges#367, Churn#41]\n",
      "   +- BroadcastHashJoin [customerID#0], [customerID#263], LeftOuter, BuildRight, false\n",
      "      :- Project [customerID#0, Churn#41, tenure#61L, TotalCharges#62, PhoneService#97, MultipleLines#101, InternetService#199, OnlineSecurity#200, OnlineBackup#201, DeviceProtection#202, TechSupport#203, StreamingTV#204, StreamingMovies#205, Contract#233, PaperlessBilling#258, PaymentMethod#239]\n",
      "      :  +- SortMergeJoin [customerID#0], [customerID#324], LeftOuter\n",
      "      :     :- Project [customerID#0, Churn#41, tenure#61L, TotalCharges#62, PhoneService#97, MultipleLines#101, InternetService#199, OnlineSecurity#200, OnlineBackup#201, DeviceProtection#202, TechSupport#203, StreamingTV#204, StreamingMovies#205]\n",
      "      :     :  +- SortMergeJoin [customerID#0], [customerID#306], LeftOuter\n",
      "      :     :     :- Project [customerID#0, Churn#41, tenure#61L, TotalCharges#62, PhoneService#97, MultipleLines#101]\n",
      "      :     :     :  +- SortMergeJoin [customerID#0], [customerID#295], LeftOuter\n",
      "      :     :     :     :- Project [customerID#0, Churn#41, tenure#61L, TotalCharges#62]\n",
      "      :     :     :     :  +- SortMergeJoin [customerID#0], [customerID#286], LeftOuter\n",
      "      :     :     :     :     :- Sort [customerID#0 ASC NULLS FIRST], false, 0\n",
      "      :     :     :     :     :  +- HashAggregate(keys=[customerID#0], functions=[])\n",
      "      :     :     :     :     :     +- Exchange hashpartitioning(customerID#0, 200), ENSURE_REQUIREMENTS, [id=#550]\n",
      "      :     :     :     :     :        +- HashAggregate(keys=[customerID#0], functions=[])\n",
      "      :     :     :     :     :           +- Project [customerID#0]\n",
      "      :     :     :     :     :              +- FileScan parquet [customerID#0,month#4] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<customerID:string>\n",
      "      :     :     :     :     +- Project [customerID#286, Churn#41, tenure#61L, TotalCharges#62]\n",
      "      :     :     :     :        +- SortMergeJoin [customerID#286], [customerID#66], Inner\n",
      "      :     :     :     :           :- Project [customerID#286, isnotnull(Churn#30) AS Churn#41]\n",
      "      :     :     :     :           :  +- SortMergeJoin [customerID#286], [Churn#30], LeftOuter\n",
      "      :     :     :     :           :     :- Sort [customerID#286 ASC NULLS FIRST], false, 0\n",
      "      :     :     :     :           :     :  +- HashAggregate(keys=[customerID#286], functions=[])\n",
      "      :     :     :     :           :     :     +- Exchange hashpartitioning(customerID#286, 200), ENSURE_REQUIREMENTS, [id=#552]\n",
      "      :     :     :     :           :     :        +- HashAggregate(keys=[customerID#286], functions=[])\n",
      "      :     :     :     :           :     :           +- Project [customerID#286]\n",
      "      :     :     :     :           :     :              +- Filter isnotnull(customerID#286)\n",
      "      :     :     :     :           :     :                 +- FileScan parquet [customerID#286,month#290] Batched: true, DataFilters: [isnotnull(customerID#286)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(customerID)], ReadSchema: struct<customerID:string>\n",
      "      :     :     :     :           :     +- Sort [Churn#30 ASC NULLS FIRST], false, 0\n",
      "      :     :     :     :           :        +- Exchange hashpartitioning(Churn#30, 200), ENSURE_REQUIREMENTS, [id=#556]\n",
      "      :     :     :     :           :           +- Project [customerID#32 AS Churn#30]\n",
      "      :     :     :     :           :              +- Filter ((isnotnull(kind#33) AND (kind#33 = AccountTermination)) AND isnotnull(customerID#32))\n",
      "      :     :     :     :           :                 +- FileScan parquet [customerID#32,kind#33,month#36] Batched: true, DataFilters: [isnotnull(kind#33), (kind#33 = AccountTermination), isnotnull(customerID#32)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(kind), EqualTo(kind,AccountTermination), IsNotNull(customerID)], ReadSchema: struct<customerID:string,kind:string>\n",
      "      :     :     :     :           +- Project [customerID#66, coalesce(event_counts#23L, 0) AS tenure#61L, coalesce(total_charges#25, 0.00) AS TotalCharges#62]\n",
      "      :     :     :     :              +- SortMergeJoin [customerID#66], [customerID#44], LeftOuter\n",
      "      :     :     :     :                 :- Sort [customerID#66 ASC NULLS FIRST], false, 0\n",
      "      :     :     :     :                 :  +- HashAggregate(keys=[customerID#66], functions=[])\n",
      "      :     :     :     :                 :     +- Exchange hashpartitioning(customerID#66, 200), ENSURE_REQUIREMENTS, [id=#561]\n",
      "      :     :     :     :                 :        +- HashAggregate(keys=[customerID#66], functions=[])\n",
      "      :     :     :     :                 :           +- Project [customerID#66]\n",
      "      :     :     :     :                 :              +- Filter isnotnull(customerID#66)\n",
      "      :     :     :     :                 :                 +- FileScan parquet [customerID#66,month#70] Batched: true, DataFilters: [isnotnull(customerID#66)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(customerID)], ReadSchema: struct<customerID:string>\n",
      "      :     :     :     :                 +- Sort [customerID#44 ASC NULLS FIRST], false, 0\n",
      "      :     :     :     :                    +- Exchange hashpartitioning(customerID#44, 200), ENSURE_REQUIREMENTS, [id=#567]\n",
      "      :     :     :     :                       +- HashAggregate(keys=[customerID#44, kind#45], functions=[count(value#46), sum(UnscaledValue(value#46))])\n",
      "      :     :     :     :                          +- Exchange hashpartitioning(customerID#44, kind#45, 200), ENSURE_REQUIREMENTS, [id=#563]\n",
      "      :     :     :     :                             +- HashAggregate(keys=[customerID#44, kind#45], functions=[partial_count(value#46), partial_sum(UnscaledValue(value#46))])\n",
      "      :     :     :     :                                +- Project [customerID#44, kind#45, value#46]\n",
      "      :     :     :     :                                   +- Filter ((isnotnull(kind#45) AND (kind#45 = Charge)) AND isnotnull(customerID#44))\n",
      "      :     :     :     :                                      +- FileScan parquet [customerID#44,kind#45,value#46,month#48] Batched: true, DataFilters: [isnotnull(kind#45), (kind#45 = Charge), isnotnull(customerID#44)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(kind), EqualTo(kind,Charge), IsNotNull(customerID)], ReadSchema: struct<customerID:string,kind:string,value:decimal(8,2)>\n",
      "      :     :     :     +- Sort [customerID#295 ASC NULLS FIRST], false, 0\n",
      "      :     :     :        +- Project [customerID#295, CASE WHEN isnull(PhoneService#82) THEN No ELSE Yes END AS PhoneService#97, CASE WHEN isnull(PhoneService#82) THEN No phone service ELSE CASE WHEN isnull(MultipleLines#85) THEN No ELSE Yes END END AS MultipleLines#101]\n",
      "      :     :     :           +- BroadcastHashJoin [customerID#295], [customerID#91], LeftOuter, BuildRight, false\n",
      "      :     :     :              :- Project [customerID#295, PhoneService#82]\n",
      "      :     :     :              :  +- BroadcastHashJoin [customerID#295], [customerID#76], LeftOuter, BuildRight, false\n",
      "      :     :     :              :     :- HashAggregate(keys=[customerID#295], functions=[])\n",
      "      :     :     :              :     :  +- Exchange hashpartitioning(customerID#295, 200), ENSURE_REQUIREMENTS, [id=#580]\n",
      "      :     :     :              :     :     +- HashAggregate(keys=[customerID#295], functions=[])\n",
      "      :     :     :              :     :        +- Project [customerID#295]\n",
      "      :     :     :              :     :           +- Filter isnotnull(customerID#295)\n",
      "      :     :     :              :     :              +- FileScan parquet [customerID#295,month#299] Batched: true, DataFilters: [isnotnull(customerID#295)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(customerID)], ReadSchema: struct<customerID:string>\n",
      "      :     :     :              :     +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]),false), [id=#583]\n",
      "      :     :     :              :        +- Project [customerID#76, Yes AS PhoneService#82]\n",
      "      :     :     :              :           +- Filter ((isnotnull(feature#77) AND (feature#77 = PhoneService)) AND isnotnull(customerID#76))\n",
      "      :     :     :              :              +- FileScan parquet [customerID#76,feature#77] Batched: true, DataFilters: [isnotnull(feature#77), (feature#77 = PhoneService), isnotnull(customerID#76)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(feature), EqualTo(feature,PhoneService), IsNotNull(customerID)], ReadSchema: struct<customerID:string,feature:string>\n",
      "      :     :     :              +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]),false), [id=#587]\n",
      "      :     :     :                 +- Project [customerID#91, Yes AS MultipleLines#85]\n",
      "      :     :     :                    +- Filter ((isnotnull(feature#92) AND (feature#92 = MultipleLines)) AND isnotnull(customerID#91))\n",
      "      :     :     :                       +- FileScan parquet [customerID#91,feature#92] Batched: true, DataFilters: [isnotnull(feature#92), (feature#92 = MultipleLines), isnotnull(customerID#91)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(feature), EqualTo(feature,MultipleLines), IsNotNull(customerID)], ReadSchema: struct<customerID:string,feature:string>\n",
      "      :     :     +- Sort [customerID#306 ASC NULLS FIRST], false, 0\n",
      "      :     :        +- Project [customerID#306, CASE WHEN isnull(InternetService#124) THEN No ELSE InternetService#124 END AS InternetService#199, CASE WHEN ((InternetService#124 = No) OR isnull(InternetService#124)) THEN No internet service ELSE CASE WHEN isnull(OnlineSecurity#127) THEN No ELSE OnlineSecurity#127 END END AS OnlineSecurity#200, CASE WHEN ((InternetService#124 = No) OR isnull(InternetService#124)) THEN No internet service ELSE CASE WHEN isnull(OnlineBackup#130) THEN No ELSE OnlineBackup#130 END END AS OnlineBackup#201, CASE WHEN ((InternetService#124 = No) OR isnull(InternetService#124)) THEN No internet service ELSE CASE WHEN isnull(DeviceProtection#133) THEN No ELSE DeviceProtection#133 END END AS DeviceProtection#202, CASE WHEN ((InternetService#124 = No) OR isnull(InternetService#124)) THEN No internet service ELSE CASE WHEN isnull(TechSupport#136) THEN No ELSE TechSupport#136 END END AS TechSupport#203, CASE WHEN ((InternetService#124 = No) OR isnull(InternetService#124)) THEN No internet service ELSE CASE WHEN isnull(StreamingTV#139) THEN No ELSE StreamingTV#139 END END AS StreamingTV#204, CASE WHEN ((InternetService#124 = No) OR isnull(InternetService#124)) THEN No internet service ELSE CASE WHEN isnull(StreamingMovies#142) THEN No ELSE StreamingMovies#142 END END AS StreamingMovies#205]\n",
      "      :     :           +- BroadcastHashJoin [customerID#306], [customerID#188], LeftOuter, BuildRight, false\n",
      "      :     :              :- Project [customerID#306, InternetService#124, OnlineSecurity#127, OnlineBackup#130, DeviceProtection#133, TechSupport#136, StreamingTV#139]\n",
      "      :     :              :  +- BroadcastHashJoin [customerID#306], [customerID#178], LeftOuter, BuildRight, false\n",
      "      :     :              :     :- Project [customerID#306, InternetService#124, OnlineSecurity#127, OnlineBackup#130, DeviceProtection#133, TechSupport#136]\n",
      "      :     :              :     :  +- BroadcastHashJoin [customerID#306], [customerID#169], LeftOuter, BuildRight, false\n",
      "      :     :              :     :     :- Project [customerID#306, InternetService#124, OnlineSecurity#127, OnlineBackup#130, DeviceProtection#133]\n",
      "      :     :              :     :     :  +- BroadcastHashJoin [customerID#306], [customerID#161], LeftOuter, BuildRight, false\n",
      "      :     :              :     :     :     :- Project [customerID#306, InternetService#124, OnlineSecurity#127, OnlineBackup#130]\n",
      "      :     :              :     :     :     :  +- BroadcastHashJoin [customerID#306], [customerID#154], LeftOuter, BuildRight, false\n",
      "      :     :              :     :     :     :     :- Project [customerID#306, InternetService#124, OnlineSecurity#127]\n",
      "      :     :              :     :     :     :     :  +- BroadcastHashJoin [customerID#306], [customerID#148], LeftOuter, BuildRight, false\n",
      "      :     :              :     :     :     :     :     :- Project [customerID#306, InternetService#124]\n",
      "      :     :              :     :     :     :     :     :  +- BroadcastHashJoin [customerID#306], [customerID#105], LeftOuter, BuildRight, false\n",
      "      :     :              :     :     :     :     :     :     :- HashAggregate(keys=[customerID#306], functions=[])\n",
      "      :     :              :     :     :     :     :     :     :  +- Exchange hashpartitioning(customerID#306, 200), ENSURE_REQUIREMENTS, [id=#595]\n",
      "      :     :              :     :     :     :     :     :     :     +- HashAggregate(keys=[customerID#306], functions=[])\n",
      "      :     :              :     :     :     :     :     :     :        +- Project [customerID#306]\n",
      "      :     :              :     :     :     :     :     :     :           +- Filter isnotnull(customerID#306)\n",
      "      :     :              :     :     :     :     :     :     :              +- FileScan parquet [customerID#306,month#310] Batched: true, DataFilters: [isnotnull(customerID#306)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(customerID)], ReadSchema: struct<customerID:string>\n",
      "      :     :              :     :     :     :     :     :     +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]),false), [id=#598]\n",
      "      :     :              :     :     :     :     :     :        +- Project [customerID#105, value#107 AS InternetService#124]\n",
      "      :     :              :     :     :     :     :     :           +- Filter ((isnotnull(feature#106) AND (feature#106 = InternetService)) AND isnotnull(customerID#105))\n",
      "      :     :              :     :     :     :     :     :              +- FileScan parquet [customerID#105,feature#106,value#107] Batched: true, DataFilters: [isnotnull(feature#106), (feature#106 = InternetService), isnotnull(customerID#105)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(feature), EqualTo(feature,InternetService), IsNotNull(customerID)], ReadSchema: struct<customerID:string,feature:string,value:string>\n",
      "      :     :              :     :     :     :     :     +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]),false), [id=#602]\n",
      "      :     :              :     :     :     :     :        +- Project [customerID#148, value#150 AS OnlineSecurity#127]\n",
      "      :     :              :     :     :     :     :           +- Filter ((isnotnull(feature#149) AND (feature#149 = OnlineSecurity)) AND isnotnull(customerID#148))\n",
      "      :     :              :     :     :     :     :              +- FileScan parquet [customerID#148,feature#149,value#150] Batched: true, DataFilters: [isnotnull(feature#149), (feature#149 = OnlineSecurity), isnotnull(customerID#148)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(feature), EqualTo(feature,OnlineSecurity), IsNotNull(customerID)], ReadSchema: struct<customerID:string,feature:string,value:string>\n",
      "      :     :              :     :     :     :     +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]),false), [id=#606]\n",
      "      :     :              :     :     :     :        +- Project [customerID#154, value#156 AS OnlineBackup#130]\n",
      "      :     :              :     :     :     :           +- Filter ((isnotnull(feature#155) AND (feature#155 = OnlineBackup)) AND isnotnull(customerID#154))\n",
      "      :     :              :     :     :     :              +- FileScan parquet [customerID#154,feature#155,value#156] Batched: true, DataFilters: [isnotnull(feature#155), (feature#155 = OnlineBackup), isnotnull(customerID#154)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(feature), EqualTo(feature,OnlineBackup), IsNotNull(customerID)], ReadSchema: struct<customerID:string,feature:string,value:string>\n",
      "      :     :              :     :     :     +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]),false), [id=#610]\n",
      "      :     :              :     :     :        +- Project [customerID#161, value#163 AS DeviceProtection#133]\n",
      "      :     :              :     :     :           +- Filter ((isnotnull(feature#162) AND (feature#162 = DeviceProtection)) AND isnotnull(customerID#161))\n",
      "      :     :              :     :     :              +- FileScan parquet [customerID#161,feature#162,value#163] Batched: true, DataFilters: [isnotnull(feature#162), (feature#162 = DeviceProtection), isnotnull(customerID#161)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(feature), EqualTo(feature,DeviceProtection), IsNotNull(customerID)], ReadSchema: struct<customerID:string,feature:string,value:string>\n",
      "      :     :              :     :     +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]),false), [id=#614]\n",
      "      :     :              :     :        +- Project [customerID#169, value#171 AS TechSupport#136]\n",
      "      :     :              :     :           +- Filter ((isnotnull(feature#170) AND (feature#170 = TechSupport)) AND isnotnull(customerID#169))\n",
      "      :     :              :     :              +- FileScan parquet [customerID#169,feature#170,value#171] Batched: true, DataFilters: [isnotnull(feature#170), (feature#170 = TechSupport), isnotnull(customerID#169)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(feature), EqualTo(feature,TechSupport), IsNotNull(customerID)], ReadSchema: struct<customerID:string,feature:string,value:string>\n",
      "      :     :              :     +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]),false), [id=#618]\n",
      "      :     :              :        +- Project [customerID#178, value#180 AS StreamingTV#139]\n",
      "      :     :              :           +- Filter ((isnotnull(feature#179) AND (feature#179 = StreamingTV)) AND isnotnull(customerID#178))\n",
      "      :     :              :              +- FileScan parquet [customerID#178,feature#179,value#180] Batched: true, DataFilters: [isnotnull(feature#179), (feature#179 = StreamingTV), isnotnull(customerID#178)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(feature), EqualTo(feature,StreamingTV), IsNotNull(customerID)], ReadSchema: struct<customerID:string,feature:string,value:string>\n",
      "      :     :              +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]),false), [id=#622]\n",
      "      :     :                 +- Project [customerID#188, value#190 AS StreamingMovies#142]\n",
      "      :     :                    +- Filter ((isnotnull(feature#189) AND (feature#189 = StreamingMovies)) AND isnotnull(customerID#188))\n",
      "      :     :                       +- FileScan parquet [customerID#188,feature#189,value#190] Batched: true, DataFilters: [isnotnull(feature#189), (feature#189 = StreamingMovies), isnotnull(customerID#188)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(feature), EqualTo(feature,StreamingMovies), IsNotNull(customerID)], ReadSchema: struct<customerID:string,feature:string,value:string>\n",
      "      :     +- Sort [customerID#324 ASC NULLS FIRST], false, 0\n",
      "      :        +- Project [customerID#324, Contract#233, CASE WHEN isnull(PaperlessBilling#236) THEN No ELSE PaperlessBilling#236 END AS PaperlessBilling#258, PaymentMethod#239]\n",
      "      :           +- BroadcastHashJoin [customerID#324], [customerID#251], LeftOuter, BuildRight, false\n",
      "      :              :- Project [customerID#324, Contract#233, PaperlessBilling#236]\n",
      "      :              :  +- BroadcastHashJoin [customerID#324], [customerID#245], LeftOuter, BuildRight, false\n",
      "      :              :     :- Project [customerID#324, Contract#233]\n",
      "      :              :     :  +- BroadcastHashJoin [customerID#324], [customerID#214], LeftOuter, BuildRight, false\n",
      "      :              :     :     :- HashAggregate(keys=[customerID#324], functions=[])\n",
      "      :              :     :     :  +- Exchange hashpartitioning(customerID#324, 200), ENSURE_REQUIREMENTS, [id=#630]\n",
      "      :              :     :     :     +- HashAggregate(keys=[customerID#324], functions=[])\n",
      "      :              :     :     :        +- Project [customerID#324]\n",
      "      :              :     :     :           +- Filter isnotnull(customerID#324)\n",
      "      :              :     :     :              +- FileScan parquet [customerID#324,month#328] Batched: true, DataFilters: [isnotnull(customerID#324)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(customerID)], ReadSchema: struct<customerID:string>\n",
      "      :              :     :     +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]),false), [id=#633]\n",
      "      :              :     :        +- Project [customerID#214, value#216 AS Contract#233]\n",
      "      :              :     :           +- Filter ((isnotnull(feature#215) AND (feature#215 = Contract)) AND isnotnull(customerID#214))\n",
      "      :              :     :              +- FileScan parquet [customerID#214,feature#215,value#216] Batched: true, DataFilters: [isnotnull(feature#215), (feature#215 = Contract), isnotnull(customerID#214)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(feature), EqualTo(feature,Contract), IsNotNull(customerID)], ReadSchema: struct<customerID:string,feature:string,value:string>\n",
      "      :              :     +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]),false), [id=#637]\n",
      "      :              :        +- Project [customerID#245, value#247 AS PaperlessBilling#236]\n",
      "      :              :           +- Filter ((isnotnull(feature#246) AND (feature#246 = PaperlessBilling)) AND isnotnull(customerID#245))\n",
      "      :              :              +- FileScan parquet [customerID#245,feature#246,value#247] Batched: true, DataFilters: [isnotnull(feature#246), (feature#246 = PaperlessBilling), isnotnull(customerID#245)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(feature), EqualTo(feature,PaperlessBilling), IsNotNull(customerID)], ReadSchema: struct<customerID:string,feature:string,value:string>\n",
      "      :              +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]),false), [id=#641]\n",
      "      :                 +- Project [customerID#251, value#253 AS PaymentMethod#239]\n",
      "      :                    +- Filter ((isnotnull(feature#252) AND (feature#252 = PaymentMethod)) AND isnotnull(customerID#251))\n",
      "      :                       +- FileScan parquet [customerID#251,feature#252,value#253] Batched: true, DataFilters: [isnotnull(feature#252), (feature#252 = PaymentMethod), isnotnull(customerID#251)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(feature), EqualTo(feature,PaymentMethod), IsNotNull(customerID)], ReadSchema: struct<customerID:string,feature:string,value:string>\n",
      "      +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]),false), [id=#650]\n",
      "         +- Project [customerID#263, ((year(cast(now#270 as date)) > (year(dateOfBirth#264) + 65)) OR ((year(cast(now#270 as date)) = (year(dateOfBirth#264) + 65)) AND ((month(cast(now#270 as date)) < month(dateOfBirth#264)) OR ((month(cast(now#270 as date)) = month(dateOfBirth#264)) AND (dayofmonth(cast(now#270 as date)) <= dayofmonth(cast(now#270 as date))))))) AS SeniorCitizen#279, Partner#267, Dependents#268, gender#265, MonthlyCharges#269]\n",
      "            +- Filter isnotnull(customerID#263)\n",
      "               +- FileScan parquet [customerID#263,dateOfBirth#264,gender#265,Partner#267,Dependents#268,MonthlyCharges#269,now#270] Batched: true, DataFilters: [isnotnull(customerID#263)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/home/yuanli/work/customer-churn/data-science-blueprints/chu..., PartitionFilters: [], PushedFilters: [IsNotNull(customerID)], ReadSchema: struct<customerID:string,dateOfBirth:date,gender:string,Partner:string,Dependents:string,MonthlyC...\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "wide_data.explain()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 33:======================================================> (28 + 1) / 29]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 1.15 s, sys: 188 ms, total: 1.34 s\n",
      "Wall time: 2min 58s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "from churn.etl import write_df\n",
    "write_df(wide_data, output_file)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Inspecting individual tables\n",
    "\n",
    "If we need to inspect individual components of our processing, we can.  Each constituent of these joins is registered as a temporary view.  For example, we loaded `customers` earlier using a method from `churn.etl`, but it is also available as a table:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "customers = session.table(\"customers\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2022-04-05 10:02:56,112 WARN rapids.GpuOverrides: \n",
      "!Exec <CollectLimitExec> cannot run on GPU because the Exec CollectLimitExec has been disabled, and is disabled by default because Collect Limit replacement can be slower on the GPU, if huge number of rows in a batch it could help by limiting the number of rows transferred from GPU to CPU. Set spark.rapids.sql.exec.CollectLimitExec to true if you wish to enable it\n",
      "  @Partitioning <SinglePartition$> could run on GPU\n",
      "\n",
      "2022-04-05 10:02:56,113 WARN rapids.GpuOverrides: \n",
      "!Exec <CollectLimitExec> cannot run on GPU because the Exec CollectLimitExec has been disabled, and is disabled by default because Collect Limit replacement can be slower on the GPU, if huge number of rows in a batch it could help by limiting the number of rows transferred from GPU to CPU. Set spark.rapids.sql.exec.CollectLimitExec to true if you wish to enable it\n",
      "  @Partitioning <SinglePartition$> could run on GPU\n",
      "\n",
      "2022-04-05 10:02:56,114 WARN rapids.GpuOverrides: \n",
      "!Exec <CollectLimitExec> cannot run on GPU because the Exec CollectLimitExec has been disabled, and is disabled by default because Collect Limit replacement can be slower on the GPU, if huge number of rows in a batch it could help by limiting the number of rows transferred from GPU to CPU. Set spark.rapids.sql.exec.CollectLimitExec to true if you wish to enable it\n",
      "  @Partitioning <SinglePartition$> could run on GPU\n",
      "\n",
      "2022-04-05 10:03:35,633 WARN rapids.GpuOverrides: =============>(790 + 1) / 795]\n",
      "!Exec <CollectLimitExec> cannot run on GPU because the Exec CollectLimitExec has been disabled, and is disabled by default because Collect Limit replacement can be slower on the GPU, if huge number of rows in a batch it could help by limiting the number of rows transferred from GPU to CPU. Set spark.rapids.sql.exec.CollectLimitExec to true if you wish to enable it\n",
      "  @Partitioning <SinglePartition$> could run on GPU\n",
      "\n",
      "2022-04-05 10:03:35,634 WARN rapids.GpuOverrides: \n",
      "!Exec <CollectLimitExec> cannot run on GPU because the Exec CollectLimitExec has been disabled, and is disabled by default because Collect Limit replacement can be slower on the GPU, if huge number of rows in a batch it could help by limiting the number of rows transferred from GPU to CPU. Set spark.rapids.sql.exec.CollectLimitExec to true if you wish to enable it\n",
      "  @Partitioning <SinglePartition$> could run on GPU\n",
      "\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|          customerID|\n",
      "+--------------------+\n",
      "|9102-OXKFY-fCyaBG...|\n",
      "|5478-JJVZK-pUoKfE...|\n",
      "|1843-TLSGD-PjdWrt...|\n",
      "|2027-FECZV-5HMwOd...|\n",
      "|3793-MMFUH-FBa4QK...|\n",
      "|5360-XGYAZ-F1ALBc...|\n",
      "|1843-TLSGD-L@JxWt...|\n",
      "|5872-OEQNH-5NXyac...|\n",
      "|6773-LQTVT-XB@vuC...|\n",
      "|3301-VKTGC-PjdWrt...|\n",
      "|9251-AWQGT-fCyaBG...|\n",
      "|9830-ECLEN-lqhKlh...|\n",
      "|7969-FFOWG-fPARzA...|\n",
      "|9451-WLYRI-0stTDJ...|\n",
      "|4293-ETKAP-dkh3P1...|\n",
      "|6281-FKEWS-0V3zMQ...|\n",
      "|8220-OCUFY-PjdWrt...|\n",
      "|0578-SKVMF-GSLp0h...|\n",
      "|2165-VOEGB-K8kBya...|\n",
      "|6754-WKSHP-rt81Nn...|\n",
      "+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "customers.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can see which tables are available by querying the session catalog:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2022-04-05 10:03:38,813 WARN conf.HiveConf: HiveConf of name hive.stats.jdbc.timeout does not exist\n",
      "2022-04-05 10:03:38,814 WARN conf.HiveConf: HiveConf of name hive.stats.retries.wait does not exist\n",
      "2022-04-05 10:03:40,550 WARN metastore.ObjectStore: Version information not found in metastore. hive.metastore.schema.verification is not enabled so recording the schema version 2.3.0\n",
      "2022-04-05 10:03:40,550 WARN metastore.ObjectStore: setMetaStoreSchemaVersion called but recording version is disabled: version = 2.3.0, comment = Set by MetaStore yuanli@127.0.1.1\n",
      "2022-04-05 10:03:40,703 WARN metastore.ObjectStore: Failed to get database global_temp, returning NoSuchObjectException\n",
      "2022-04-05 10:03:40,833 WARN rapids.GpuOverrides: \n",
      "! <LocalTableScanExec> cannot run on GPU because GPU does not currently support the operator class org.apache.spark.sql.execution.LocalTableScanExec\n",
      "  @Expression <AttributeReference> name#507 could run on GPU\n",
      "  @Expression <AttributeReference> database#508 could run on GPU\n",
      "  @Expression <AttributeReference> description#509 could run on GPU\n",
      "  @Expression <AttributeReference> tableType#510 could run on GPU\n",
      "  @Expression <AttributeReference> isTemporary#511 could run on GPU\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "['churned',\n",
       " 'contracts',\n",
       " 'counts_and_charges',\n",
       " 'customer_account_features',\n",
       " 'customer_account_meta',\n",
       " 'customer_billing',\n",
       " 'customer_charges',\n",
       " 'customer_internet_features',\n",
       " 'customer_phone_features',\n",
       " 'customers',\n",
       " 'device_protection',\n",
       " 'internet_service',\n",
       " 'multiple_lines',\n",
       " 'online_backup',\n",
       " 'online_security',\n",
       " 'paperless',\n",
       " 'payment',\n",
       " 'phone_service',\n",
       " 'streaming_movies',\n",
       " 'streaming_tv',\n",
       " 'tech_support',\n",
       " 'terminations']"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tables = session.catalog.listTables()\n",
    "[t.name for t in tables]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Finishing up"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "session.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
